package com.goldmantis.activemq.ch5;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;

import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMapMessage;

public class Publisher2 {
	
    private int MAX_DELTA_PERCENT = 1;
    private Map<String, Double> LAST_PRICES = new HashMap<String, Double>();
    private static int count = 1;
    
    private static String brokerURL = "tcp://localhost:51616";
    private static transient ConnectionFactory factory;
    private transient TopicConnection connection;
    private transient TopicSession session;
    private transient TopicPublisher producer;
    
    private String username = "publisher";
    private String password = "password";
    
    public Publisher2() throws JMSException {
    	factory = new ActiveMQConnectionFactory(brokerURL);
    	connection = (TopicConnection)factory.createConnection(username, password);
        connection.start();
        session = (TopicSession)connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }
    
    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }
    
    public static void main(String[] args) throws JMSException {
    	Publisher2 publisher = new Publisher2();
    	TopicSession session = publisher.getSession();
    	Topic destination = session.createTopic("STOCKS.topic");
    	MessageProducer producer =(TopicPublisher)session.createPublisher(destination);
    	InputStream ins= System.in;
    	InputStreamReader reader = new InputStreamReader(ins);
    	BufferedReader br=new BufferedReader(reader); 
    	boolean jump = false;
    	while(!jump){
    		try {
    			System.out.println("请输入要发送的信息");
				String msg = br.readLine();
	            Message message = session.createTextMessage(msg);
	            producer.send(destination, message);
	            System.out.println("Published price messages");					
			} catch (IOException e) {
				e.printStackTrace();
			}	
    	}
        publisher.close();
    }

    public void sendMessage(String[] stocks) throws JMSException {
        int idx = 0;
        while (true) {
            idx = (int)Math.round(stocks.length * Math.random());
            if (idx < stocks.length) {
                break;
            }
        }
        String stock = stocks[idx];
        Destination destination = session.createTopic("STOCKS." + stock);
        Message message = createStockMessage(stock, session);
        System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);
        producer.send(destination, message);
    }

    protected Message createStockMessage(String stock, Session session) throws JMSException {
        Double value = LAST_PRICES.get(stock);
        if (value == null) {
            value = new Double(Math.random() * 100);
        }

        // lets mutate the value by some percentage
        double oldPrice = value.doubleValue();
        value = new Double(mutatePrice(oldPrice));
        LAST_PRICES.put(stock, value);
        double price = value.doubleValue();

        double offer = price * 1.001;

        boolean up = (price > oldPrice);
        MapMessage message = session.createMapMessage();
        message.setString("stock", stock);
        message.setDouble("price", price);
        message.setDouble("offer", offer);
        message.setBoolean("up", up);
        return message;
    }

    protected double mutatePrice(double price) {
        double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;
        return price * (100 + percentChange) / 100;
    }
    
    public TopicSession getSession(){
    	return this.session;
    }
    
    public MessageProducer getProducer(){
    	return this.producer;
    }
}
